---
title: Asset Catalog | Dagster
description: Dagster provides a data-aware orchestration graph.
---

# Asset Catalog

"Asset" is Dagster's word for an entity, external to solids, that is mutated or created by solids.
An asset might be a table in a database that solids append to, an ML model in a model store that
solids overwrite, or even a slack channel that solids write messages to.

## Relevant APIs

| Name                                                        | Description                                                      |
| ----------------------------------------------------------- | ---------------------------------------------------------------- |
| <PyObject module="dagster" object="AssetMaterialization" /> | Dagster event indicating that a solid has materialized an asset. |

## Overview

The Asset Catalog is an interface inside [Dagit](/concepts/dagit/dagit) that centers on assets. Each entry
in the catalog is an asset and includes:

- Runs that mutated or created the asset.
- Metadata logged by developers about the asset.

Developers place entries in the Asset Catalog by yielding <PyObject module="dagster" object="AssetMaterialization" />
events from inside their solids or [output managers](/concepts/io-management/io-managers). The act
of mutating or creating an asset is called a "materialization". <PyObject module="dagster" object="AssetMaterialization" />
events may include arbitrary metadata that describes the asset at the time of the materialization.

There are two general patterns for dealing with assets when using Dagster:

- Materialize assets from inside the body of a solid.
- Focus solid on pure business logic, and delegate the materialization of assets to IOManagers.

---

## Yielding an AssetMaterialization

To make the Asset Catalog aware that we materialized an asset in our solid, we can yield an <PyObject module="dagster" object="AssetMaterialization" /> event.
This would involve changing the following solid:

```python file=/concepts/assets/materialization_solids.py startafter=start_materialization_solids_marker_0 endbefore=end_materialization_solids_marker_0
@solid
def my_simple_solid(_):
    df = read_df()
    remote_storage_path = persist_to_storage(df)
    return remote_storage_path
```

into something like this:

```python file=/concepts/assets/materialization_solids.py startafter=start_materialization_solids_marker_1 endbefore=end_materialization_solids_marker_1
@solid
def my_materialization_solid(context):
    df = read_df()
    remote_storage_path = persist_to_storage(df)
    yield AssetMaterialization(asset_key="my_dataset", description="Persisted result to storage")
    yield Output(remote_storage_path)
```

Note: Our materialization solid must now explicitly yield an <PyObject module="dagster" object="Output" />
event instead of relying on the implicit conversion of the return value into an <PyObject module="dagster" object="Output" />
event.

We should now see a materialization event in the event log when we execute this solid.

<!-- This was generated with:
    * `dagit -f materialization_solids.py -p 3333` inside docs_snippets/concepts/assets yuhan/docs-asset/ directory
-->

<img src="/images/concepts/assets/asset-materialization.png" />

## Asset Catalog in Dagit

After emitting an AssetMaterialization event, we should now also see an entry in the Asset Catalog.

Note: The Asset Catalog in Dagit is only enabled for instances configured with an "asset-aware"
event log storage, such as `PostgresEventLogStorage` and `ConsolidatedSqliteEventLogStorage`. Here
is an example of an asset-aware instance configuration:

```yaml file=/concepts/assets/dagster.yaml
run_launcher:
  module: dagster.core.launcher
  class: DefaultRunLauncher

event_log_storage:
  module: dagster.core.storage.event_log
  class: ConsolidatedSqliteEventLogStorage
  config:
    base_dir: "/var/shared/dagster_home"
```

_Please refer to the [Dagster Instance](/deployment/dagster-instance) section for more information_
_about configuring your Dagster instance._

Once the Dagster instance is configured with an asset-aware event log storage, after you run a
pipeline which generates an asset, you'll find the asset listed on the `Assets` tab in Dagit.

<img src="/images/concepts/assets/asset-page.png" />

You can also find the details about it by clicking the asset key of it, `my_dataset` in this case.

<img src="/images/concepts/assets/asset-details.png" />

Clicking the \[View Asset Dashboard] on the materialization event from the log generated above can
also bring you to the corresponding Asset details page.

## Examples

### Recording Asset Materializations in Output Managers

To record that an OutputManager or IOManager has mutated or created an asset, we can yield an <PyObject module="dagster" object="AssetMaterialization" /> event from its `handle_output` method.

```python file=/concepts/assets/materialization_io_managers.py startafter=start_marker_0 endbefore=end_marker_0
class PandasCsvIOManager(IOManager):
    def load_input(self, context):
        file_path = os.path.join(["my_base_dir", context.step_key, context.output_name])
        return pd.read_csv(file_path)

    def handle_output(self, context, obj):
        file_path = os.path.join(["my_base_dir", context.step_key, context.output_name])

        obj.to_csv(file_path)

        yield AssetMaterialization(
            asset_key=AssetKey(file_path), description="Persisted result to storage."
        )
```

### Attaching Metadata to the Asset Materialization

There are a variety of types of metadata that can be associated with a materialization event, all
through the <PyObject module="dagster" object="EventMetadataEntry" /> class. Each materialization
event optionally takes a list of metadata entries that are then displayed in the event log and the
Asset Catalog.

Example with a solid:

```python file=concepts/assets/materialization_solids.py startafter=start_materialization_solids_marker_2 endbefore=end_materialization_solids_marker_2
@solid
def my_metadata_materialization_solid(context):
    df = read_df()
    remote_storage_path = persist_to_storage(df)
    yield AssetMaterialization(
        asset_key="my_dataset",
        description="Persisted result to storage",
        metadata_entries=[
            EventMetadataEntry.text("Text-based metadata for this event", label="text_metadata"),
            EventMetadataEntry.fspath(remote_storage_path),
            EventMetadataEntry.url("http://mycoolsite.com/url_for_my_data", label="dashboard_url"),
            EventMetadataEntry.float(calculate_bytes(df), "size (bytes)"),
        ],
    )
    yield Output(remote_storage_path)
```

Example with an IOManager:

```python file=concepts/assets/materialization_io_managers.py startafter=start_marker_1 endbefore=end_marker_1
class PandasCsvIOManagerWithMetadata(IOManager):
    def load_input(self, context):
        file_path = os.path.join(["my_base_dir", context.step_key, context.output_name])
        return pd.read_csv(file_path)

    def handle_output(self, context, obj):
        file_path = os.path.join(["my_base_dir", context.step_key, context.output_name])

        obj.to_csv(file_path)

        yield AssetMaterialization(
            asset_key=AssetKey(file_path),
            description="Persisted result to storage.",
            metadata_entries=[
                EventMetadataEntry.int(obj.shape[0], label="number of rows"),
                EventMetadataEntry.float(obj["some_column"].mean(), "some_column mean"),
            ],
        )
```

Check our API docs for <PyObject module="dagster" object="EventMetadataEntry" /> for more details
on they types of event metadata available.
